package org.geektimes.reactive.streams;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

/**
 * Created with IntelliJ IDEA.
 * Description:
 *
 * @Author: menglinggang
 * @Date: 2021-03-31
 * @Time: 3:24 下午
 */
public class DefaultSubscription<T> implements Subscription, Subscriber<T> {

    private boolean cancel;

    private Long request;

    private Long count = 0L;

    private Subscriber<T> subscriber;

    public DefaultSubscription(Subscriber subscriber) {
        this.subscriber = subscriber;
    }

    @Override
    public void request(long n) {
        if(Long.MAX_VALUE < n) {
            subscriber.onError(new RuntimeException("request is over Long.MAX_VALUE"));
        }
        request = n;
        count = 0L;
    }

    @Override
    public void cancel() {
        cancel = true;
    }

    public Subscriber getSubscriber() {
        return subscriber;
    }

    public boolean isCancel() {
        return cancel;
    }

    @Override
    public void onSubscribe(Subscription s) {
        subscriber.onSubscribe(this);
    }

    @Override
    public void onNext(T t) {
        if (request == null) {
            return ;
        }
        count ++;
        // 执行
        subscriber.onNext(t);
    }

    @Override
    public void onError(Throwable t) {
        subscriber.onError(t);
    }

    @Override
    public void onComplete() {
        if (count >= request) subscriber.onComplete();
    }
}
